package com.swak.http.retry;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.http.builder.PendingConfirm;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.timer.HashedWheelTimer;
import com.swak.timer.Timeout;
import com.swak.timer.Timer;
import com.swak.timer.TimerTask;

/**
 * 执行失败才加入重试服务，默认 2秒 的重试时间
 * 
 * @author lifeng
 * @date 2021年3月11日 下午3:20:21
 */
public class RetryService {

	private Logger logger = LoggerFactory.getLogger(RetryService.class);
	private final ConcurrentMap<String, RetryTask> failedPosted = new ConcurrentHashMap<>();
	private final HashedWheelTimer retryTimer;
	private long retryPeriod = 2 * 1000;

	public RetryService() {
		retryTimer = new HashedWheelTimer(new SwakThreadFactory("HttpRetryTimer", true, new AtomicInteger(1)),
				retryPeriod, TimeUnit.MILLISECONDS, 128);
	}

	@SuppressWarnings("unchecked")
	public <T> CompletableFuture<T> add(PendingConfirm pendingConfirm) {
		RetryTask oldOne = failedPosted.get(pendingConfirm.getId());
		if (oldOne != null) {
			return (CompletableFuture<T>) oldOne.future;
		}
		RetryTask newTask = new RetryTask(pendingConfirm);
		oldOne = failedPosted.putIfAbsent(pendingConfirm.getId(), newTask);
		if (oldOne == null) {
			pendingConfirm.doCall().whenComplete((r, e) -> {
				if (e == null) {
					this.doDel(pendingConfirm.getId());
					newTask.future.complete((T) r);
				} else {
					logger.error("Http 请求异常，加入重试, 最大重试次数", pendingConfirm.getRetryTimes(), e);
					retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
				}
			});
			return newTask.future;
		}
		return (CompletableFuture<T>) oldOne.future;
	}

	private void doDel(String id) {
		RetryTask f = failedPosted.remove(id);
		if (f != null) {
			f.cancel();
		}
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public class RetryTask implements TimerTask {

		private final PendingConfirm pendingConfirm;
		private final int retryTimes;
		private volatile int times = 1;
		private volatile boolean cancel;
		private CompletableFuture future;

		public RetryTask(PendingConfirm pendingConfirm) {
			this.pendingConfirm = pendingConfirm;
			this.retryTimes = pendingConfirm.getRetryTimes();
			this.cancel = false;
			this.future = new CompletableFuture<>();
		}

		public void cancel() {
			this.cancel = true;
		}

		public boolean isCancel() {
			return cancel;
		}

		@Override
		public void run(Timeout timeout) throws Exception {
			if (timeout.isCancelled() || timeout.timer().isStop() || isCancel()) {
				if (!this.future.isDone()) {
					this.future.completeExceptionally(new TimeoutException("超过最大的重试次数！最大重试次数 " + retryTimes));
				}
				return;
			}
			if (times > retryTimes) {
				if (!this.future.isDone()) {
					this.future.completeExceptionally(new TimeoutException("超过最大的重试次数！最大重试次数 " + retryTimes));
				}
				return;
			}
			try {
				doRetry(timeout).whenComplete((r, e) -> {
					if (e != null) {
						if (logger.isDebugEnabled()) {
							logger.debug("Http 请求异常，重试次数 {}, 最大重试次数 ", times, retryTimes, e);
						}
						this.reput(timeout, retryPeriod);
					} else {
						this.future.complete(r);
					}
				});
			} catch (Throwable t) {
				reput(timeout, retryPeriod);
			}
		}

		protected void reput(Timeout timeout, long tick) {
			if (timeout == null) {
				throw new IllegalArgumentException();
			}

			Timer timer = timeout.timer();
			if (timer.isStop() || timeout.isCancelled() || isCancel()) {
				return;
			}
			times++;
			timer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS);
		}

		protected CompletableFuture<Object> doRetry(Timeout timeout) {
			failedPosted.remove(pendingConfirm.getId());
			return this.pendingConfirm.doCall();
		}
	}
}
